/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package org.apache.skywalking.apm.agent.core.sampling;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.skywalking.apm.agent.core.boot.BootService;
import org.apache.skywalking.apm.agent.core.boot.DefaultImplementor;
import org.apache.skywalking.apm.agent.core.boot.DefaultNamedThreadFactory;
import org.apache.skywalking.apm.agent.core.boot.ServiceManager;
import org.apache.skywalking.apm.agent.core.conf.Config;
import org.apache.skywalking.apm.agent.core.conf.dynamic.ConfigurationDiscoveryService;
import org.apache.skywalking.apm.agent.core.conf.dynamic.watcher.SamplingRateWatcher;
import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment;
import org.apache.skywalking.apm.agent.core.logging.api.ILog;
import org.apache.skywalking.apm.agent.core.logging.api.LogManager;
import org.apache.skywalking.apm.util.RunnableWithExceptionProtection;

/**
 * The <code>SamplingService</code> take charge of how to sample the {@link TraceSegment}. Every {@link TraceSegment}s
 * have been traced, but, considering CPU cost of serialization/deserialization, and network bandwidth, the agent do NOT
 * send all of them to collector, if SAMPLING is on.
 * <p>
 * By default, SAMPLING is on, and  {@link Config.Agent#SAMPLE_N_PER_3_SECS }
 *
 * 采样服务
 *
 * SamplingService负责如何对TraceSegment进行采样。每个TraceSegment都已被跟踪，但是，考虑到序列化/反序列化的 CPU 成本和网络带宽，如果 SAMPLING 打开，代理不会将它们全部发送到收集器。
 * 默认情况下，SAMPLING 开启， Config.Agent.SAMPLE_N_PER_3_SECS采集样服务。
 *
 */
@DefaultImplementor
public class SamplingService implements BootService {
    private static final ILog LOGGER = LogManager.getLogger(SamplingService.class);

    // 是否运行
    private volatile boolean on = false;

    /**
     * 累加3秒内已经采样的次数
     */
    private volatile AtomicInteger samplingFactorHolder;

    /**
     * 每三秒重置一次 samplingFactorHolder
     */
    private volatile ScheduledFuture<?> scheduledFuture;

    /**
     *针对采样率这个配置，进行watcher监听
     */
    private SamplingRateWatcher samplingRateWatcher;

    @Override
    public void prepare() {
    }

    @Override
    public void boot() {
        // 初始化 watcher
        samplingRateWatcher = new SamplingRateWatcher("agent.sample_n_per_3_secs", this);

        // 注册监听
        // 当服务端下发 采样率 配置变更的时候。会通过到 samplingRateWatcher.notify()
        // samplingRateWatcher.notify()  会将数据传递到 this.handleSamplingRateChanged()
        ServiceManager.INSTANCE.findService(ConfigurationDiscoveryService.class)
                               .registerAgentConfigChangeWatcher(samplingRateWatcher);

        // 处理采样率更改
        // 为什么在boot 中调用这个方法？
        // 因为把初始化逻辑也放在 handleSamplingRateChanged 中了。
        handleSamplingRateChanged();
    }

    @Override
    public void onComplete() {

    }

    @Override
    public void shutdown() {
        if (scheduledFuture != null) {
            scheduledFuture.cancel(true);
        }
    }

    /**
     * 8.9.0 之前，这个方法的注释都是过时的
     *
     * 准确的描述是 如果采样机制没有开启， 即 on = false， 那么就表示每一条采集到的链路都会上报给 OAP
     *
     *
     * @param operationName The first operation name of the new tracing context.
     * @return true, if sampling mechanism is on, and getDefault the sampling factor successfully.
     */
    public boolean trySampling(String operationName) {
        if (on) {
            // 开启采样，根据配置的采样率 来决定 链路数据是否上报

            int factor = samplingFactorHolder.get();
            if (factor < samplingRateWatcher.getSamplingRate()) {
                return samplingFactorHolder.compareAndSet(factor, factor + 1);
            } else {
                return false;
            }
        }
        return true;
    }

    /**
     * 强制采样
     *
     * Increase the sampling factor by force, to avoid sampling too many traces. If many distributed traces require
     * sampled, the trace beginning at local, has less chance to be sampled.
     */
    public void forceSampled() {
        if (on) {
            samplingFactorHolder.incrementAndGet();
        }
    }

    private void resetSamplingFactor() {
        samplingFactorHolder = new AtomicInteger(0);
    }

    /**
     * 处理 采样率变更
     * Handle the samplingRate changed.
     */
    public void handleSamplingRateChanged() {
        // 获取当前采样率，如果大于0，根据配置的采样率进行采样
        if (samplingRateWatcher.getSamplingRate() > 0) {
            // 如果没工作
            if (!on) {
                on = true;
                // 重置 `累加3秒内已经采样的次数` 为 0
                this.resetSamplingFactor();

                // 每3秒重置一次 `累加3秒内已经采样的次数`   resetSamplingFactor

                ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor(
                    new DefaultNamedThreadFactory("SamplingService"));
                scheduledFuture = service.scheduleAtFixedRate(new RunnableWithExceptionProtection(
                    this::resetSamplingFactor, t -> LOGGER.error("unexpected exception.", t)), 0, 3, TimeUnit.SECONDS);
                LOGGER.debug(
                    "Agent sampling mechanism started. Sample {} traces in 3 seconds.",
                    samplingRateWatcher.getSamplingRate()
                );
            }
        } else {
            // 不开启 采样。 就给他关了
            if (on) {
                if (scheduledFuture != null) {
                    scheduledFuture.cancel(true);
                }
                on = false;
            }
        }
    }
}
